package com.zpl.mq.consumer.api;


import com.zpl.mq.consumer.constant.ConsumerRespCode;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import lombok.extern.slf4j.Slf4j;

/**
 * @author zhupanlin
 * @version 1.0
 * @description: TODO
 * @date 2024/9/7 17:35
 */
@Slf4j
public class MqConsumerPush extends Thread implements IMqConsumer{

    @Override
    public void run() {
      log.info("MQ 消费者开始启动服务端 groupName: {}, port: {}, brokerAddress: {}", 
              groupName, port, brokerAddress);

        EventLoopGroup boosGroup = new NioEventLoopGroup();
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        
        try {
            ServerBootstrap serverBootstrap = new ServerBootstrap();
            serverBootstrap.group(workerGroup, boosGroup)
                    .channel(NioServerSocketChannel.class)
                    // 添加处理器，处理器里面是真正处理业务的
                    .childHandler(new ChannelInitializer<Channel>() {
                        @Override
                        protected void initChannel(Channel channel) throws Exception {
                            channel.pipeline().addLast(new MqConsumerHanlder());
                        }
                    })
                    // 这个参数影响的是还没有被accept 取出的连接 因没有空闲线程导致连接请求被暂存放到队列中的长度
                    .option(ChannelOption.SO_BACKLOG, 128)
                    // 这个参数只是过一段时间内客户端没有响应，服务端会发送一个 ack 包，以判断客户端是否还活着
                    .childOption(ChannelOption.SO_KEEPALIVE, true);
            // 绑定端口，开始接收进来的链接
            ChannelFuture channelFuture = serverBootstrap.bind(port).syncUninterruptibly();
            log.info("MQ 消费者启动完成，监听【" + port + "】端口");
            channelFuture.channel().closeFuture().syncUninterruptibly();
            log.info("MQ 消费者关闭完成");
        } catch (Exception e) {
            log.error("MQ 消费者启动异常", e);
            throw new MQException(ConsumerRespCode.RPC_INIT_FAILED);
        } finally {
            workerGroup.shutdownGracefully();
            boosGroup.shutdownGracefully();
        }
    }
}
